/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package bgu.sonar.util.sync;

import bgu.sonar.util.ex.UncheckedInterruptedException;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 *
 * @author bennyl
 */
public class ProducerConsumerExecutorService implements ExecutorService {

    public static ProducerConsumerExecutorService wrap(ExecutorService pool, int max) {
        return new ProducerConsumerExecutorService(pool, max);
    }

    public static ProducerConsumerExecutorService rewrap(ProducerConsumerExecutorService pool, int workers) {
        return wrap(pool.wrapee, workers);
    }
    
    private ExecutorService wrapee;
    private Semaphore maxGuard;
    private int max;

    public ProducerConsumerExecutorService(ExecutorService wrapee, int maxQueue) {
        this.wrapee = wrapee;
        maxGuard = new Semaphore(maxQueue);
        this.max = maxQueue;
    }

    @Override
    public void shutdown() {
        maxGuard.release(max);
        wrapee.shutdown();
    }

    /**
     * @return null!
     */
    @Override
    public List<Runnable> shutdownNow() {
        maxGuard.release(max);
        wrapee.shutdownNow();
        return null;
    }

    @Override
    public boolean isShutdown() {
        return wrapee.isShutdown();
    }

    @Override
    public boolean isTerminated() {
        return wrapee.isTerminated();
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return wrapee.awaitTermination(timeout, unit);
    }

    @Override
    public <T> Future<T> submit(final Callable<T> task) {
        try {
            maxGuard.acquire();
            return wrapee.submit(new Callable<T>() {
                @Override
                public T call() throws Exception {
                    try {
                        return task.call();
                    } finally {
                        maxGuard.release();
                    }
                }
            });
        } catch (InterruptedException ex) {
            throw new UncheckedInterruptedException(ex);
        }
    }

    @Override
    public <T> Future<T> submit(final Runnable task, final T result) {
        try {
            maxGuard.acquire();
            return wrapee.submit(new Callable<T>() {
                @Override
                public T call() throws Exception {
                    try {
                        task.run();
                        return result;
                    } finally {
                        maxGuard.release();
                    }
                }
            });
        } catch (InterruptedException ex) {
            throw new UncheckedInterruptedException(ex);
        }
    }

    @Override
    public Future<?> submit(final Runnable task) {
        try {
            maxGuard.acquire();
            return wrapee.submit(new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    try {
                        task.run();
                        return null;
                    } finally {
                        maxGuard.release();
                    }
                }
            });
        } catch (InterruptedException ex) {
            throw new UncheckedInterruptedException(ex);
        }
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
        throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
        throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
        throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
    }

    @Override
    public void execute(final Runnable command) {
        try {
            maxGuard.acquire();
            wrapee.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        command.run();
                    } finally {
                        maxGuard.release();
                    }
                }
            });
        } catch (InterruptedException ex) {
            throw new UncheckedInterruptedException(ex);
        }
    }

    /**
     * will wait until all requested processes in this queue is done
     */
    public void awaitAllProcessed() throws InterruptedException {
        final Semaphore await = new Semaphore(0);
        execute(new Runnable(){

            @Override
            public void run() {
                await.release();
            }
            
        });
        
        await.acquire();
    }
}
